package com.yanggu.flink.datastream_api.state.keyedstate

import cn.hutool.core.date.DateUtil
import com.yanggu.flink.datastream_api.pojo.Event
import com.yanggu.flink.datastream_api.source.ClickSource
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.common.state.{ReducingState, ReducingStateDescriptor, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.util.Date

/**
 * 每隔10s, 输出一次每个客户最后一次访问详情
 */
object ReducingStateDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    env
      .addSource(new ClickSource)
      .assignAscendingTimestamps(data => data.getTimestamp)
      .keyBy(data => data.getName)
      .process(new LastEventFunction)
      .print()

    env.execute()

  }

}

class LastEventFunction extends KeyedProcessFunction[String, Event, String] {

  private var reducingState: ReducingState[Event] = _

  private lazy val timestampState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timestamp", classOf[Long]))

  override def open(parameters: Configuration): Unit = {
    val reducingStateDescriptor = new ReducingStateDescriptor[Event]("reducing", new ReduceFunction[Event] {
      override def reduce(value1: Event, value2: Event) = value2
    }, classOf[Event])

    reducingState = getRuntimeContext.getReducingState(reducingStateDescriptor)

  }

  override def processElement(value: Event, ctx: KeyedProcessFunction[String, Event, String]#Context, out: Collector[String]): Unit = {
    reducingState.add(value)

    //如果没有注册定时器, 就注册
    if (timestampState.value() == 0) {
      val timestamp = value.getTimestamp + 10000L
      ctx.timerService().registerEventTimeTimer(timestamp)
      timestampState.update(timestamp)
    }

  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, Event, String]#OnTimerContext, out: Collector[String]): Unit = {
    val event = reducingState.get()
    out.collect(s"用户: ${event.getName}, 在10s内, 最后一次, 访问url: ${event.getUrl}, 访问日期: ${DateUtil.formatDateTime(new Date(event.getTimestamp))}")
    timestampState.clear()
    reducingState.clear()
  }

}